package Autodesk;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import Autodesk.LookForNotFound.LogMapper;
import Autodesk.LookForNotFound.LogReducer;

public class LookForNotFoundHaProxy extends Configured implements Tool {
	
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		private static final String EXCLUDE_VALUES = "beehive_monitor";  
		private static final String EXCLUDE_STYLES = "/styles/";  
		private static final String EXCLUDE_JS = ".js";  
		private static Logger logger = Logger.getLogger(ParseLogs.class);


		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		private Map<ArrayList<String>,String> StaticListLang=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListProd=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListChannel=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListFeature=new HashMap<ArrayList<String>,String>();
		private ArrayList<String> listChannelsExcluded=new ArrayList<String>();
		Parser par=new Parser();
		
		public void setup(Context context){
			Configuration conf=context.getConfiguration();
//			 if (conf.getBoolean("language.patterns", false) | conf.getBoolean("product.patterns", false) | conf.getBoolean("feature.patterns", false) |
//					 conf.getBoolean("channel.patterns", false) | conf.getBoolean("version.patterns", false) | conf.getBoolean("time.patterns", false) ){
//				  Path[] patternsFiles = new Path[0];
//				  try{
//			           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
//				  }catch(IOException ioe){
//			           System.err.println("Caught exception while getting cached files: ");
//				  }
//				  for (Path patternsFile : patternsFiles) {
//					  
//					  	if( (conf.get("language.path")).equals(patternsFile.getName())){
//					  		par.getStaticList(patternsFile,StaticListLang,"lang"); 
//					  	}else if((conf.get("product.path").equals(patternsFile.getName()))){
//					  		par.getStaticList(patternsFile,StaticListProd,"products"); 
////					  	}else if((conf.get("feature.path").equals(patternsFile.getName()))){
////					  		par.getStaticList(patternsFile,StaticListFeature,"features"); 
//					  	}else if((conf.get("channel.path").equals(patternsFile.getName()))){
//					  		par.getStaticList(patternsFile,StaticListChannel,"channel"); 
//					  	}
//				  }
//				 
//			 }
			StaticListLang=par.getStaticList("language"); 
			logger.info("Loaded Langs-------------------");
	  		
	  		StaticListProd=par.getStaticList("product"); 
	  		logger.info("Loaded Products-------------------");
	  		
	  		StaticListChannel=par.getStaticList("channel"); 
	  		logger.info("Loaded Channels-------------------");
	  		
			 inicitializeListChannelsExc();

		}
		private void inicitializeListChannelsExc() {
			listChannelsExcluded.add("productchannel");
			 listChannelsExcluded.add("productchannel2");
			 listChannelsExcluded.add("supportchannel");
			 listChannelsExcluded.add("supportchannel2");
			 listChannelsExcluded.add("acadconnectchannel");
			 listChannelsExcluded.add("acadconnectchannel2");
			 listChannelsExcluded.add("taxonomy3");
		}
			
	private String generateKeyProducts(String queryline,String tokenProd, String tokenRele) {
		String tokenSub;
		if(queryline.indexOf("origin=upi")<0 ){//IT IS A BASELINE VALUE
			tokenSub=tokenProd+"\t"+""+"\t"+tokenRele;
		}else{
			tokenSub=""+"\t"+tokenProd+"\t"+tokenRele;
		}
		return tokenSub;
	}
	
	
	//MAP FUNCTION
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

		//String line = value.toString().toLowerCase();
		String line = value.toString();
		String channel = null;

		// Looking for CHANNEL.
		String channelID;
		channel = par.getChannel(line);

		channelID= par.lookForKey(StaticListChannel, channel.toLowerCase(),"channel");
		if ( (channelID.equals("1")) && !(listChannelsExcluded.contains(channel))){
			word.set("channels"+"\t"+channel);
		}
		if( (line.indexOf(EXCLUDE_STYLES) > -1) && (line.indexOf(EXCLUDE_JS)> -1) && listChannelsExcluded.contains(channel) && line.indexOf(EXCLUDE_VALUES)>-1){				
			// skip
		} else if(line.indexOf("?")>-1){
			String queryLine=line.substring(line.indexOf('?')+1);
		 	StringTokenizer tokenizer = new StringTokenizer(queryLine,"&");
			while (tokenizer.hasMoreTokens()) {
				String token = tokenizer.nextToken();							
					if (token.indexOf("p=") > -1) {
						try{
							String prodAb=par.getProduct(token, queryLine,par);								
							String[] productRele=prodAb.split("\t");
							String tokenProd=productRele[0];
							String tokenRele=productRele[1];
							String id=par.lookForKeyProduct(StaticListProd, tokenProd.toLowerCase(),tokenRele.toLowerCase());
							if(id.equals("1") && !tokenProd.equals("null")){
								id=generateKeyProducts(queryLine, tokenProd, tokenRele);
								word.set("products"+"\t"+id);
							}
						}catch(Exception e){
						}
					} else if ((token.indexOf("l=") > -1) && (token.indexOf("channel=") < 0) && (token.indexOf("url=") < 0)) {		
						String tokenSub = token.substring(2);
						if(par.lookForKey(StaticListLang, tokenSub.toLowerCase(),"languages").equals("1")){
							word.set("languages"+"\t"+tokenSub);
						}								
					}
				
				context.write(word, one);
			}
		}

	}

}

//REDUCE FUNCTION	
public static class LogReducer extends
		Reducer<Text, IntWritable, Text, IntWritable> {
	
	private IntWritable result = new IntWritable();
	 private MultipleOutputs<Text,IntWritable> mos;
	 public void setup(Context context) {
	 mos = new MultipleOutputs<Text,IntWritable>(context);
	 }
	public void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
		
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		result.set(sum);
		String[] recordIdentifier=key.toString().split("\t",2);
		if(recordIdentifier[0].equals("products")){
			mos.write("ProductsNotFound", new Text(recordIdentifier[1]), result);
		}else if(recordIdentifier[0].equals("languages")){
			mos.write("LanguagesNotFound", new Text(recordIdentifier[1]), result);
		}else if(recordIdentifier[0].equals("channels")){
			mos.write("ChannelsNotFound", new Text(recordIdentifier[1]), result);
//		}else if(recordIdentifier[0].equals("features")){
//			mos.write("FeaturesNotFound", new Text(recordIdentifier[1]), result);
		}else{
			System.out.println("ERROR: "+key.toString());
		}
		context.write(key, result);
	}
	 public void cleanup(Context context) throws IOException, InterruptedException {
		 mos.close();
		 }
}

public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "ParseLogs");
	job.setJarByClass(Autodesk.LookForNotFound.class);
	
//	FileInputFormat.addInputPath(job, new Path(args[0]));
//	FileOutputFormat.setOutputPath(job, new Path(args[1]));
	
	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);
	
	List<String> other_args = new ArrayList<String>();
	for (int i = 0; i < args.length; ++i) {
		other_args.add(args[i]);
	}
//	for (int i = 0; i < args.length; ++i) {
//		if ("-languages".equals(args[i])) {
//			prodPath=new Path(args[++i]);
//			DistributedCache.addCacheFile(prodPath.toUri(),
//					job.getConfiguration());
//			job.getConfiguration().setBoolean("language.patterns",
//					true);
//			job.getConfiguration().set("language.path", prodPath.getName());
//		} else if ("-products".equals(args[i])) {
//			prodPath=new Path(args[++i]);
//			DistributedCache.addCacheFile(prodPath.toUri(),
//					job.getConfiguration());
//			job.getConfiguration().setBoolean("product.patterns",
//					true);
//			job.getConfiguration().set("product.path", prodPath.getName());
//		} else if ("-channels".equals(args[i])) {
//			prodPath=new Path(args[++i]);
//			DistributedCache.addCacheFile(prodPath.toUri(),
//					job.getConfiguration());
//			job.getConfiguration().setBoolean("channel.patterns",
//					true);
//			job.getConfiguration().set("channel.path", prodPath.getName());
//		} else if ("-features".equals(args[i])) {
//			prodPath=new Path(args[++i]);
//			DistributedCache.addCacheFile(prodPath.toUri(),
//					job.getConfiguration());
//			job.getConfiguration().setBoolean("feature.patterns",
//					true);
//			job.getConfiguration().set("feature.path", prodPath.getName());
//		}else{
//			other_args.add(args[i]);
//		}
//	}
    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
    
    MultipleOutputs.addNamedOutput(job, "ProductsNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
    MultipleOutputs.addNamedOutput(job, "LanguagesNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
    MultipleOutputs.addNamedOutput(job, "ChannelsNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
    MultipleOutputs.addNamedOutput(job, "FeaturesNotFound", TextOutputFormat.class, Text.class, IntWritable.class);


	System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}

public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new LookForNotFoundHaProxy(), args);
	System.exit(res);
	}
}
